package com.test.imcmqtt.mqtt2;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration;

import com.test.imcmqtt.mqtt1.PushCallback;

import java.util.LinkedList;

/**
 * MqttClient 连接池
 * 
 * @author 周西栋
 * @date
 * @param
 * @return
 */
public class MqttClientPool {
	public static Logger log = LoggerFactory.getLogger(MqttClientPool.class);
	private static MqttClientPool MQTTCLIENTPOOL = new MqttClientPool();

	private MqttClientPool() {
	}

	public static MqttClientPool instance() {
		return MQTTCLIENTPOOL;
	}

	// 连接池的最大容量
	// @Value("${mqttclientpool.maximal}")
	private static int maximal = 50;

	// 连接池的最小容量
	// @Value("${mqttclientpool.minimum}")
	private static int minimum = 20;

	// 连接池的初始容量
	// @Value("${mqttclientpool.initial}")
	private static int initial = 20;

	// 连接池的扩充容量
	// @Value("${mqttclientpool.step}")
	private static int step = 5;

	// 空闲连接池
	private static LinkedList<MqttClient> pool;

	// 已用连接池
	private static LinkedList<MqttClient> pool_used;

	// 静态代码块
	static {
		pool = new LinkedList<>();
		pool_used = new LinkedList<>();
		AutoMqttProperties properties = new AutoMqttProperties();
		MqttClientFactory factory = new MqttClientFactory();
		// 初始化连接池
		for (int i = 0; i < minimum; i++) {
			pool.add(factory.createMqttClient(properties));
		}
	}

	// 扩容
	private boolean dilatation() {
		if (pool.size() >= maximal) {
			return false;
		}
		AutoMqttProperties properties = new AutoMqttProperties();
		MqttClientFactory factory = new MqttClientFactory();
		int limitnum = (pool.size() + step > maximal) ? (pool.size() + step - maximal) : step;
		// 扩容连接池
		for (int i = 0; i < limitnum; i++) {
			pool.add(factory.createMqttClient(properties));
		}
		log.info("连接池中现在还有{}个连接", pool.size());
		return true;
	}

	// 对外提供连接
	public MqttClient getClient() {
		MqttClient client = null;
		if (pool.size() > minimum || pool.size() > 0) {
			client = pool.get(0);
			pool.remove(0);
			pool_used.add(client);
		} else if (pool.size() <= minimum || pool.size() == 0) {
			dilatation();
			client = pool.get(0);
			pool.remove(0);
			pool_used.add(client);
		} else {
			log.error("实在抱歉，我是mqtt连接池，我已经没有连接可以提供了！");
			throw new RuntimeException("实在抱歉，连接池里已经没有连接了！");
		}
		return client;
	}

	// 归还连接
	public boolean close(MqttClient client) {
		if (client == null) {
			return false;
		}
		boolean b = pool_used.remove(client);
		if (b) {
			if (pool.add(client)) {
				return true;
			} else {
				pool_used.add(client);
				return false;
			}
		} else {
			return false;
		}
	}

	public static MqttClientPool getMQTTCLIENTPOOL() {
		return MQTTCLIENTPOOL;
	}

	public static void setMQTTCLIENTPOOL(MqttClientPool mQTTCLIENTPOOL) {
		MQTTCLIENTPOOL = mQTTCLIENTPOOL;
	}

	public static int getMaximal() {
		return maximal;
	}

	public static void setMaximal(int maximal) {
		MqttClientPool.maximal = maximal;
	}

	public static int getMinimum() {
		return minimum;
	}

	public static void setMinimum(int minimum) {
		MqttClientPool.minimum = minimum;
	}

	public static int getInitial() {
		return initial;
	}

	public static void setInitial(int initial) {
		MqttClientPool.initial = initial;
	}

	public static int getStep() {
		return step;
	}

	public static void setStep(int step) {
		MqttClientPool.step = step;
	}

	public static LinkedList<MqttClient> getPool() {
		return pool;
	}

	public static void setPool(LinkedList<MqttClient> pool) {
		MqttClientPool.pool = pool;
	}

	public static LinkedList<MqttClient> getPool_used() {
		return pool_used;
	}

	public static void setPool_used(LinkedList<MqttClient> pool_used) {
		MqttClientPool.pool_used = pool_used;
	}

}
